package com.lvmama.rhino.analyze.processing

import com.lvmama.rhino.common.entity.{PENDING_PAYMENT, REGISTER, StatLogType, TerminalType}
import com.lvmama.rhino.common.utils.JDBCUtil.Conversions.ImplicitInsert
import com.lvmama.rhino.common.utils.Utils
import com.lvmama.rhino.common.utils.logging.LazyLogging
import org.apache.spark.sql.Dataset

/**
  * Created by yuanxiaofeng on 2017/6/1.
  */
class WirelessV2ChannelRelate extends Serializable with LazyLogging {
  self =>

  val yesterday = Utils.getYesterday()

  import spray.json._
  import DefaultJsonProtocol._

  def process(ds: Dataset[StatLogType]) = {
    import ds.sparkSession.implicits._

    val channelProcessed = ds
      .transform(tf =>
        tf.map(s => {
          if (s.platformCode == TerminalType.wab.toString || s.pageTypeCode == TerminalType.pc.toString) {
            s.copy(secondChannel = s.channelCode)
          } else {
            s
          }
        })
      )

    //订单分析
    val orderNumAnalyze = channelProcessed
      .filter(f => f.pageTypeCode == PENDING_PAYMENT.code)
      .groupByKey(g => (g.useYears, g.categoryId, g.platformCode))
      .mapGroups { (key, iter) =>
        val _iter = iter.map(i => (i.secondChannel, i.paymentAmount)).toList
        (key, _iter.size, _iter.map(_._2).sum,
          _iter.groupBy(g => g._1).map(m => (m._1, m._2.size)),
          _iter.groupBy(g => g._1).map(m => (m._1, m._2.map(_._2).sum)))
      }
      .map(m => (m._1._1, m._1._2, m._1._3, m._2, m._3, m._4.toJson.toString(), m._5.toJson.toString(), yesterday))
      .toDF("user_use_years", "category_id", "platform_code", "order_amount", "order_money", "order_amount_json",
        "order_money_json", "oper_date")

    orderNumAnalyze.insertDF2MysqlDirect("flow_stat_order")

    //新注册用户
    val newRegister = channelProcessed
      .filter(n => n.pageTypeCode == REGISTER.code && n.buttonCode == "register")
      .map(n => (n.platformCode, n.secondChannel))
      .groupByKey(g => (g._1, g._2))
      .count()
      .map(m => (m._1._1, m._1._2, m._2, yesterday))
      .toDF("app_type", "channel", "count", "oper_date")

    newRegister.insertDF2MysqlDirect("flow_app_new_register")
  }
}
